package edu.csl.study.kafka.demo;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 如果提交位移前崩溃，那么会导致重复消费。
 * 不管怎么编码，消费者都有这问题，一般需要额外处理。
 */
public class KafkaConsumerDemo {
  private static AtomicBoolean running = new AtomicBoolean(true);
  public static void main(String[] args) {
    manualCommitOffSet_Async();
  }
  /**
   * 自动提交偏移量：
   * - 默认是自动提交，每5秒自动提交一次位移。
   * - 提交位移：提交的位移 = 最后消费的偏移量 + 1
   * - 自动提交可能导致重复消费和数据丢失的情况。
   *    重复消费：拉取数据消费完后，正要提交位移时，数据崩溃了。下次消费时会重新拉取上次未提交的消息。
   *    数据丢失：拉取了两次数据放在BlockingQueue中，两次数据都已自动提交了，
   *   ​          但是处理线程还在处理第一批数据，如果此时处理线程挂了，那么就会丢失消息。
   * ​   注意：都是在异常情况下才会出现重复消费和数据丢失现象，正常情况下是没问题的。
   */
  public static void autoCommitOffSet(){
    //准备配置属性
    Properties props = new Properties();
    //kafka集群地址
    props.put("bootstrap.servers", "centos20:9092,centos21:9092,centos22:9092");
    //消费者组id
    props.put("group.id", "test10");
    //自动提交偏移量
    props.put("enable.auto.commit", "true");
    //自动提交偏移量的时间间隔 1秒
    props.put("auto.commit.interval.ms", "1000");
    //默认是latest
    //earliest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
    //latest:   当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
    //none :    当各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
    props.put("auto.offset.reset","earliest");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    org.apache.kafka.clients.consumer.KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
    //指定消费哪些topic
    String topic = "order";
    consumer.subscribe(Arrays.asList(topic));
    System.out.println("启动成功->");
    while (true) {
      //指定每个多久拉取一次数据
      ConsumerRecords<String, String> records = consumer.poll(1000);
      /***
       * 注意：实际消费数据时，不会这样处理，而是启动一个线程去处理。
       */
      for (ConsumerRecord<String, String> record : records)
        System.out.printf(" partion = %s, offset = %d, key = %s, value = %s%n", record.partition(),record.offset(), record.key(), record.value());
    }

  }
 public static KafkaConsumer<String, String> getManualCommitOffSetConsumer(){
   Properties props = new Properties();
   props.put("bootstrap.servers", "centos20:9092,centos21:9092,centos22:9092");
   props.put("group.id", "controllerOffset");
   // 关闭自动提交，改为手动提交偏移量
   props.put("enable.auto.commit", "false");
   props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
   props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
   //拦截器
   props.put("interceptor.classes",ConsumerInterceptorTTL.class.getName());

   KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
   //指定消费者要消费的topic
   String topic = "order";
   consumer.subscribe(Arrays.asList(topic));
   return consumer;
 }
  /**
   * 手动提交位移-同步 （批量处理 + 批量提交方式）
   * 按分区粒度消费 和 批量提交都是推荐的方法
   */
  public static void manualCommitOffSet_Sync_Bath(){

    KafkaConsumer<String, String> consumer = getManualCommitOffSetConsumer();

    //定义一个数字，表示消息达到多少后手动提交偏移量
    final int minBatchSize = 20;

    //定义一个数组，缓冲一批数据
    List<ConsumerRecord<String, String>> buffer = new ArrayList<ConsumerRecord<String, String>>();
    while (running.get()) {
      ConsumerRecords<String, String> records = consumer.poll(100);
      for (ConsumerRecord<String, String> record : records) {
        buffer.add(record);
      }
      if (buffer.size() >= minBatchSize) {
        //insertIntoDb(buffer);  拿到数据之后，进行消费
        System.out.println("缓冲区的数据条数："+buffer.size());
        System.out.println("我已经处理完这一批数据了...");
        consumer.commitSync();
        buffer.clear();
      }
    }

  }
  /**
   * 同步方式 - 一条消息消费提交一次位移
   * 注意：性能太差，生产不会使用！！！
   * 缺点：如果提交位移前崩溃，那么会导致重复消费。
   *
   */
  public static  void manualCommitOffSet_Sync_Single(){
    KafkaConsumer<String, String> consumer = getManualCommitOffSetConsumer();
    int i = 0;
    try {
      while (running.get()) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records) {
          System.out.printf("消费消息： partion = %s, offset = %d, key = %s, value = %s%n", record.partition(),record.offset(), record.key(), record.value());
          long offset = record.offset();
          TopicPartition partition =
                  new TopicPartition(record.topic(), record.partition());
          consumer.commitSync(Collections
                  .singletonMap(partition, new OffsetAndMetadata(offset + 1)));
          System.out.printf("提交位移：当前位移：%d ；提交位移：%d%n",offset,offset+1);
        }
      }

    } finally {
      consumer.close();
    }

  }
  /**
   * 同步方式 - 按分区粒度消费
   *  按分区粒度消费 和 批量提交都是推荐的方法
   */
  public static void manualCommitOffSet_Sync_patition( ){
    KafkaConsumer<String, String> consumer = getManualCommitOffSetConsumer();
    try {
      while (running.get()) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (TopicPartition partition : records.partitions()) {
          List<ConsumerRecord<String, String>> partitionRecords =
                  records.records(partition);
          for (ConsumerRecord<String, String> record : partitionRecords) {
            System.out.printf("消费消息： partion = %s, offset = %d, key = %s, value = %s%n", record.partition(),record.offset(), record.key(), record.value());
          }
          long offsets = partitionRecords
                  .get(partitionRecords.size() - 1).offset();
          consumer.commitSync(Collections.singletonMap(partition,
                  new OffsetAndMetadata(offsets + 1)));
          System.out.printf("分区一次性提交位移：当前位移：%d ；提交位移：%d%n",offsets,offsets+1);
        }
      }
    } finally {
      consumer.close();
    }

  }
  /**
   * 异步发送:效率比同步发送高。
   * 小小的缺点：发生故障时，重复消费的数据比同步的要多。
   *
   */
  public static void manualCommitOffSet_Async( ){
    KafkaConsumer<String, String> consumer = getManualCommitOffSetConsumer();
    try {
      while (running.get()) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records) {
          //do some logical processing.
        }
        if(records.count() > 0){
          consumer.commitAsync(new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
                                   Exception exception) {
              if (exception == null) {
                System.out.println(offsets);
              } else {
                System.err.println("fail to commit offsets "+offsets+" 异常信息:"+exception.getMessage());
              }
            }
          });
        }

      }
    } finally {
        //这里为了防止正常退出或者发生再均衡的情况，导致位移没有提交。
        //此处只是减少重复消费，而不能完全避免。
        try {
          consumer.commitSync();
        } finally {
          consumer.close();
        }
    }

  }


}
